package com.xrui.hbase.filter;

import com.google.common.base.Objects;
import com.xrui.hbase.ColumnName;
import com.xrui.hbase.DataRequest;
import com.xrui.hbase.DataRequestBuilder;
import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.codehaus.jackson.JsonNode;
import org.codehaus.jackson.node.JsonNodeFactory;
import org.codehaus.jackson.node.ObjectNode;

import java.io.IOException;

/**
 * A RowFilter that excludes rows that have no data for some column <code>columnName</code>.
 */
public final class HasColumnDataRowFilter extends RowFilter {
    /**
     * The name of the family node.
     */
    private static final String FAMILY_NODE = "family";

    /**
     * The name of the qualifier node.
     */
    private static final String QUALIFIER_NODE = "qualifier";

    /**
     * The name of the column family to check for data in.
     */
    private final String mFamily;

    /**
     * The name of the column qualifier to check for data in.
     */
    private final String mQualifier;

    /**
     * Constructs a row filter that excludes rows that have no data in <code>columnName</code>.
     *
     * @param family    The column family of interest.
     * @param qualifier The column qualifier of interest.
     */
    public HasColumnDataRowFilter(String family, String qualifier) {
        if (null == family || family.isEmpty()) {
            throw new IllegalArgumentException("family is required");
        }
        if (null == qualifier || qualifier.isEmpty()) {
            throw new IllegalArgumentException("qualifier is required");
        }
        mFamily = family;
        mQualifier = qualifier;
    }

    /**
     * Constructs a row filter that excludes rows that have no data in <code>columnName</code>.
     *
     * @param columnName The column family:qualifier of interest.
     */
    public HasColumnDataRowFilter(String columnName) {
        if (null == columnName) {
            throw new IllegalArgumentException("columnName is required");
        }

        ColumnName colName = ColumnName.create(columnName);
        if (!colName.isFullyQualified()) {
            throw new IllegalArgumentException("Cannot use an unqualified column family.");
        }

        mFamily = colName.getFamily();
        mQualifier = colName.getQualifier();
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public DataRequest getDataRequest() {
        DataRequestBuilder builder = DataRequest.builder();
        builder.newColumnsDef().add(mFamily, mQualifier);
        return builder.build();
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public Filter toHBaseFilter() throws IOException {
        // Create a filter that accepts a cell from mFamily and mQualifier only if it is
        // not an empty byte array.  Since all hbase cells have data (at the very least, the
        // schema hash), this will accept all cells that exist.
        ColumnName columnName = ColumnName.create(mFamily, mQualifier);
        SingleColumnValueFilter filter = new SingleColumnValueFilter(
            columnName.getFamilyBytes(),
            columnName.getQualifierBytes(),
            CompareOp.NOT_EQUAL,
            new byte[0]);

        // If there are no cells in mFamily and mQualifier, skip the entire row.
        filter.setFilterIfMissing(true);

        return filter;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public boolean equals(Object other) {
        if (!(other instanceof HasColumnDataRowFilter)) {
            return false;
        } else {
            final HasColumnDataRowFilter otherFilter = (HasColumnDataRowFilter) other;
            return Objects.equal(otherFilter.mFamily, mFamily)
                && Objects.equal(otherFilter.mQualifier, mQualifier);
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public int hashCode() {
        return Objects.hashCode(mFamily, mQualifier);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    protected JsonNode toJsonNode() {
        final ObjectNode root = JsonNodeFactory.instance.objectNode();
        root.put(FAMILY_NODE, mFamily);
        root.put(QUALIFIER_NODE, mQualifier);
        return root;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    protected Class<? extends RowFilterDeserializer> getDeserializerClass() {
        return HasColumnDataRowFilterDeserializer.class;
    }

    /**
     * Deserializes {@code HasColumnDataRowFilter}.
     */
    public static final class HasColumnDataRowFilterDeserializer
        implements RowFilterDeserializer {
        /**
         * {@inheritDoc}
         */
        @Override
        public RowFilter createFromJson(JsonNode root) {
            final String family = root.path(FAMILY_NODE).getTextValue();
            final String qualifier = root.path(QUALIFIER_NODE).getTextValue();
            return new HasColumnDataRowFilter(family, qualifier);
        }
    }
}
